package com.lxw.study.kafka.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Headers;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class TestListener {
    @KafkaListener(id = "11", idIsGroup = false, topics = "1123", groupId = "g2")
    public void c1(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            long offset = record.offset();
            log.info("offset = " + offset);
            String value = record.value();
            log.info("value = " + value);
            if (offset == 1L) {
                ack.nack(3000);
            } else {
                ack.acknowledge();
            }
            Thread.sleep(1000);
        } catch (Exception e) {
            log.error("失败", e);

        }
    }
    @KafkaListener(id = "22", idIsGroup = false, topics = "1123", groupId = "g2")
    public void c2(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            long offset = record.offset();
            log.info("offset = " + offset);
            String value = record.value();
            log.info("value = " + value);
//            if (offset == 1L) {
//                ack.nack(3000);
//            } else {
//                ack.acknowledge();
//            }
            // 不确定每次启动都会收到
            Thread.sleep(1000);
        } catch (Exception e) {
            log.error("失败", e);

        }
    }


}
